package com.zhang.second.day07;

import com.zhang.second.utils.OrderEvents;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @title: 订单超时检测
 * @author: zhang
 * @date: 2022/2/3 08:16
 */
public class OrderTimeOut {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<String> result = env
                .readTextFile("/Users/apple/IdeaProjects/flink_1.13/src/main/resources/OrderLog.csv")
                .map(new MapFunction<String, OrderEvents>() {
                    @Override
                    public OrderEvents map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new OrderEvents(
                                fields[0], fields[1], fields[2],
                                Long.parseLong(fields[3]) * 1000L
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderEvents>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvents>() {
                                    @Override
                                    public long extractTimestamp(OrderEvents element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })

                )
                .keyBy(r -> r.orderId)
                .process(new KeyedProcessFunction<String, OrderEvents, String>() {
                    private ValueState<OrderEvents> createState;
                    private ValueState<Long> timerTs;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        createState = getRuntimeContext().getState(
                                new ValueStateDescriptor<OrderEvents>(
                                        "create", Types.POJO(OrderEvents.class)
                                )
                        );
                        timerTs = getRuntimeContext().getState(
                                new ValueStateDescriptor<Long>(
                                        "timer", Types.LONG
                                )
                        );
                    }

                    @Override
                    public void processElement(OrderEvents value, Context ctx, Collector<String> out) throws Exception {
                        if ("create".equals(value.type)) {
                            createState.update(value);
                            //等待支付15分钟
                            long ts = value.ts + 15 * 60 * 1000L;
                            ctx.timerService().registerEventTimeTimer(ts);
                            timerTs.update(ts);
                        } else if ("pay".equals(value.type)) {
                            OrderEvents orderEvents = createState.value();
                            if (orderEvents == null) {
                                ctx.output(new OutputTag<String>("Payed TimerOut or No Create") {
                                           },
                                        value.orderId + "Payed But No Create");
                            } else {
                                out.collect(value.orderId + "Create at" + orderEvents.ts
                                        + "Payed at" + value.ts);
                                ctx.timerService().deleteEventTimeTimer(timerTs.value());
                                createState.clear();
                                timerTs.clear();
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, OrderEvents, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        OrderEvents orderEvents = createState.value();
                        ctx.output(new OutputTag<String>("No Pay") {
                                   },
                                orderEvents.orderId + "Create But No Pay!");
                        createState.clear();
                        timerTs.clear();
                    }
                });
        result.print();
        result.getSideOutput(new OutputTag<String>("Payed TimerOut or No Create") {
        }).print("Payed TimerOut");

        result.getSideOutput(new OutputTag<String>("No Pay") {
        }).print("No Pay");

        env.execute();
    }
}
